package cluster

import (
	"gitee.com/cooge/impush/cluster/client"
	"gitee.com/cooge/impush/cluster/conn"
	"gitee.com/cooge/impush/config"
	"gitee.com/cooge/impush/core"
	"gitee.com/cooge/impush/gutil"
	log "github.com/sirupsen/logrus"
	"time"
	"gitee.com/cooge/impush/util"
	"gitee.com/cooge/impush/message"
	"errors"
	"strconv"
	"sync"
	"runtime"
)

var clientMap = gutil.NewLinkExMap()

func Start() {

	go startServer(config.CLUSTER_LOCAL_PORT)
	go startClient(config.CLUSTER_REMOTE_HOST, config.CLUSTER_REMOTE_PORT)
	go Live()
	/**
	添加监听
	*/
	addUserFilter()

	/**
	   没发现的用户批量发
	 */

	core.AddMessageFilter(func(s string, i *message.Message, e error) {
		if e!=nil{
			log.Debug("未发现用户",s,"向全体集群机器发")
			conn.SendMessage(i)
		}
	})
}

func Live() {
	timer1 := time.NewTicker(time.Duration(30) * time.Second)
	for {
		select {
		case <-timer1.C:
			{
				syncMachine()
			}
		}
	}
}

func SendALLUser(clusterClient *client.ClusterClient) {
	us := core.GetALLUsers()
	urry := make([]string, 0)
	for index, k := range us {
		uu := core.GetUser(k)
		for _, vc := range uu {
			if !isClusterUser(vc) {
				urry = append(urry, k)
				break
			}
		}
		if len(urry)>0&&(index+1)>>7<<7 == (index+1)  {
			log.Info("批量发送用户",clusterClient.Raddress,":",len(urry))
			data:=SendClusterUsers(urry)
			_, err := clusterClient.Write(data)
			if err != nil {
				return
			}
			urry = make([]string, 0)
			runtime.Gosched()
		}
	}
	if len(urry) > 0 {
		log.Debug("余量发送用户",clusterClient.Raddress,":",len(urry))
		data:=SendClusterUsers(urry)
	
		_, err := clusterClient.Write(data)
		if err != nil {
			return
		}
	}
}

func dealClientConn(clusterClient *client.ClusterClient, i int) {
	if i == config.CLUSTER_CLIENT_START_SUCCESS {
		go syncMachine()
		go SendALLUser(clusterClient)

	}
	if i == config.CLUSTER_CLIENT_NET_BREAK {
		clientMap.Delete(clusterClient.Raddress, clusterClient)
	}

	if i==config.CLUSTER_CLIENT_START_ERROR{
		clientMap.Delete(clusterClient.Raddress, clusterClient)
	}
	core.AddInfo(config.CLUSTER_CONN_SERVER, clientMap.GetALLKey())
}



func dealMessage(clusterClient *client.ClusterClient,msg *message.Message,userId string)error{
	cs:=core.GetUser(userId)
	if cs!=nil{
		for _,u:=range cs{
			if !isClusterUser(u){
				_,err:=core.SendMessage(userId,msg);
				return err
			}
		}
	}
	return errors.New(message.NO_USER);
}

func dealUser(clusterClient *client.ClusterClient,userId string,status int){

	sendAddUser(userId)

}
func startClient(rhost string, rport int) {
	lock.Lock()
	defer lock.Unlock()
	lrp := config.CLUSTER_LOCAL_PORT
	log.Info("启动集群客户端",rhost,":",rport)
	if !(util.IsLocal(rhost) && len(rhost)>0 && rport == lrp) {
		var address = util.Merge(rhost,":",strconv.Itoa(rport))
		cn := client.NewClusterClient2(address)
		clientMap.Add(address, cn)
		cn.Start(dealClientConn,dealMessage,dealUser)
	} else {
		log.Info("不与自身连接",rhost,rport)
	}
}

func sendAddUser(userId ...string) {
	keys := clientMap.GetALLKey()
	if len(keys) > 0 {
		var data = SendClusterUsers(userId)
		for _, key := range keys {
			mm := clientMap.Get(key)
			for _, v := range mm {
				vv, ok := v.(*client.ClusterClient)
				if ok&&vv.Status==client.RUN {
					_, err := vv.Write(data)
					if err == nil {
						break
					}
				}
			}
		}
	}
}
func sendDeleteUser(userId string,ti *time.Time) {
	keys := clientMap.GetALLKey()
	if len(keys) > 0 {
		var data = DeleteClusterUsers(userId,ti)

		for _, key := range keys {
			mm := clientMap.Get(key)
			for _, v := range mm {
				vv, ok := v.(*client.ClusterClient)
				if ok &&vv.Status==client.RUN{
					_, err := vv.Write(data)
					if err == nil {
						break
					}
				}
			}
		}
	}
}

func syncMachine() {

	keys := clientMap.GetALLKey()
	if len(keys) > 0 {
		for _, key := range keys {
			mm := clientMap.Get(key)
			for _, v := range mm {
				vv, ok := v.(*client.ClusterClient)
				if ok && vv.Status == client.RUN{
						_, err := vv.Live(keys)
						if err == nil {
							break
						}

				}
			}
		}
	}
}

/**
建立新的连接
*/
var lock sync.Mutex
func connback(clu *conn.ClusterConn, address []string) {
	lock.Lock()
	defer lock.Unlock()
	for _, k := range address {
		if !clientMap.IsHasKey(k) {
			cn := client.NewClusterClient2(k)
			clientMap.Add(k, cn)
			cn.Start(dealClientConn,dealMessage,dealUser)
		}
	}
}

func startServer(lport int) {
	log.Info("启动集群服务端")
	ncs := NewClusterServer()
	err := ncs.Bind(lport)
	if err == nil {
		for {
			sc, err := ncs.Accept()
			if err == nil {
				sc.Start(connback)
			} else {
				log.Panic("集群服务出现系统故障", err.Error())
				break
			}
		}
	}else{
		log.Panic("集群服务启动失败", err.Error())
	}
}
